package personal.wang.job;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import personal.wang.constant.AppConstant;
import personal.wang.dic.DicInitializer;
import personal.wang.domain.TestpointInfoProto;
import personal.wang.enm.TaskStatus;
import personal.wang.fileAccess.FileAccess;
import personal.wang.producer.DataMockProducer;
import personal.wang.producer.ProducerHandler;
import personal.wang.summary.RecordSummary;
import personal.wang.template.TemplateAnalyzer;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @className: TestPoint
 * @Description:
 * @Author: wangyifei
 * @Date: 2022/6/16 16:55
 */
public class TestPointV2 extends RecordSummary implements Runnable , Serializable {
    private static Logger logger = LoggerFactory.getLogger(TestPointV2.class);
    // 分割多个语句的标识符
    private static String SEPERATOR = ";;;" ;
    private TestpointInfoProto.TestPointInfo testPointInfo ;
    // 保存统计指标的路径
    // 字段的初始化类
    private DicInitializer dicInitializer = new DicInitializer();
    // 测试数据备份的路径
    private String backupPath ;
    private String tableName ;
    private String dicPath ;
    private String tplPath;
    private TaskStatus taskStatus ;

    public void setTaskStatus(TaskStatus taskStatus) {
        this.taskStatus = taskStatus;
    }

    public TaskStatus getTaskStatus() {
        return taskStatus;
    }

    public void setTestPointInfo(TestpointInfoProto.TestPointInfo testPointInfo) {
        this.testPointInfo = testPointInfo;
    }

    public TestpointInfoProto.TestPointInfo getTestPointInfo() {
        return testPointInfo;
    }

    public Properties getProducerProperties(){
       Properties prop = new Properties();
       prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG , testPointInfo.getKafkaBootstrap());
       prop.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG , "true");
       prop.put(ProducerConfig.LINGER_MS_CONFIG , 10);
       prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG , "lz4");
       prop.put(ProducerConfig.ACKS_CONFIG , 1);
       prop.put(ProducerConfig.RETRIES_CONFIG , 3);
       // 32 M
       prop.put(ProducerConfig.BUFFER_MEMORY_CONFIG , 33554432);
       // 16 k
       prop.put(ProducerConfig.BATCH_SIZE_CONFIG , 16384);

       prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG , StringSerializer.class.getCanonicalName());
       prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getCanonicalName());

       return prop ;
    }

    public void execute() {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                testPointInfo.getProducerNumber(),
                testPointInfo.getProducerNumber(),
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingDeque<>(),
                new BasicThreadFactory.Builder()
                        .namingPattern(testPointInfo.getName())
                        .build()
        );
        List<TestPointV2> clones = new ArrayList<>();
        TestPointV2 testPoint = null ;
        for (int i = 0; i < testPointInfo.getProducerNumber(); i++) {
            testPoint = SerializationUtils.clone(this);
            clones.add(testPoint);
            threadPoolExecutor.execute(testPoint);
        }
        threadPoolExecutor.shutdown();

        try{
            threadPoolExecutor.awaitTermination(1000 , TimeUnit.SECONDS);
            Long succSum = clones.stream().map(x -> Objects.isNull(x)? 0L:x.getSuccSum()).collect(Collectors.summingLong(x -> Objects.isNull(x)?0L:x));
            Long failSum = clones.stream().map(x -> Objects.isNull(x)? 0L:x.getFailSum()).collect(Collectors.summingLong(x -> Objects.isNull(x)?0L:x));
            this.setSuccSum(succSum);
            this.setFailSum(failSum);
            logger.info("{}" ,getSummary(testPointInfo.getName() , this.backupPath));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString() {
        return "TestPoint{" +
                testPointInfo.toString() +
                '}';
    }

    @Override
    public void run() {
        // 初始化字典
        DicInitializer.loadDics(testPointInfo.getDicCntMap());
        // 加载输出模板内容
        TemplateAnalyzer templateAnalyzer = new TemplateAnalyzer("" , testPointInfo.getTplCnt());
        String analyse = templateAnalyzer.analyse();

        // 备份发送记录
        FileAccess fa = new FileAccess();
        String backup = backupPath
                + File.separator
                + Thread.currentThread().getName().replaceAll("/" , "")
                + AppConstant.UNDERLINE
                + Thread.currentThread().getId()
                + new SimpleDateFormat("yyyyMMddhhmmss").format(new Date())
                + ".txt";
        // producer 准备
//        DataMockProducer producer = null ;
//        ProducerHandler father = new ProducerHandler();
//        if("tx".equals(testPointInfo.getProducerType())){
//            producer = new ProducerHandler.TransactionProducer();
//            producer.init(testPointInfo.getKafkaBootstrap() , UUID.randomUUID().toString());
//        }else{
//            producer = new ProducerHandler.IdempotenceProducer();
//            producer.init(testPointInfo.getKafkaBootstrap() , null);
//        }
//        ((ProducerHandler)producer).setTestPoint(this);
        String cnt = null ;
        StringBuilder sb = new StringBuilder();
        int maxVal = (testPointInfo.getMsgNumber()==-1?10:testPointInfo.getMsgNumber());
        List<String> batch = new ArrayList<>(100);
        int j = 0;
        try{
            for (int i = 0; i < maxVal; ) {
                cnt = templateAnalyzer.analyse();
                cnt = cnt.replaceAll("\\s{2,}" , "");
                batch.addAll(Arrays.stream(cnt.split(";;;")).collect(Collectors.toList()));
                if(i > 0 && i% testPointInfo.getBatchSize() == 0){
//                    producer.send(batch , sb , testPointInfo.getTopicNames((j++)% testPointInfo.getTopicNamesList().size()));
//                    fa.fileWrite(backup , sb.toString() , true);
                    logger.info("{}" , batch.get(0));
                    batch.clear();
                    sb.setLength(0);
                }
                if(maxVal != -1){
                    i++;
                }
            }
            if(!batch.isEmpty()){
//                producer.send(batch , sb , testPointInfo.getTopicNames((j++)% testPointInfo.getTopicNamesList().size()));
//                fa.fileWrite(backup , sb.toString() , true);
                logger.info("{}" , batch.get(0));
                batch.clear();
            }
        }catch (Exception e){
            logger.info("{}" , e);
        }finally {
//            producer.close();
        }
    }
    private String loadTemplate(){
        String rs = null ;
        File file = new File(tplPath);
        try{
            if(file.exists() && file.isDirectory()){
                for (File listFile : file.listFiles()) {
                    if(listFile.getName().endsWith(".tpl")){
                        rs = FileUtils.readFileToString(listFile);
                    }
                }
            }
        }catch (IOException e){
            e.printStackTrace();
        }
        return rs ;
    }

    @Override
    public void record(boolean succ, int d) {
        if(succ){
            if(Objects.isNull(succSum)){
                succSum = 0L ;
            }
            succSum += (long)d;
        }else {
            if(Objects.isNull(failSum)){
                failSum = 0L ;
            }
            failSum += (long)d;
        }
    }
}
